package com.jiesone.logger.distributecleaning.listener;
/*
 *@ClassName LoggerDistributeKafkaMQService
 *@Description TODO   统计分析日志相关业务
 *@Author tangsixiang@163.com
 *@Date 2022/1/18 17:53
 *@Version 1.0
 */


import com.jiesone.logger.distributecleaning.handler.DataProcessingHandler;
import com.jiesone.logger.distributecleaning.util.DateUtil;
import com.jiesone.logger.distributecleaning.util.JsonStringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;


@Component
public class LoggerAnalysisKafkaMQConsumer {
        Logger logger= LoggerFactory.getLogger(LoggerAnalysisKafkaMQConsumer.class);




        @Autowired
   StringRedisTemplate stringRedisTemplate;

    @KafkaListener(topics ="#{'${logger-filebeat.kafkaTopicName}'.split(',')}" ,groupId = "statistical-analysis")
    public void onMessage(ConsumerRecord<String, String> record) {

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String topic = record.topic();//订阅主体
            String serviceCode= record.key();//应用日志的
            long times= record.timestamp();
            logger.debug("====topic=="+topic+"======key=="+serviceCode+"========offset =" +  record.offset());

                                if(serviceCode!=null&&!serviceCode.equals("")){


                                    /***
                                     * 日志统计业务流程
                                     */
                                    String levelValue=   JsonStringUtil.checkLoggerLevel(record.value());// 统计个应用日志等级分布数量统计
                                    String dataTime=DateUtil.stampToTime(times,"yyyy-MM-dd-HH");
                                    String hashKey=   String.format("sourceCode:[%s]:dateDay:[%s]",serviceCode,dataTime);
                                    if(levelValue==null){
                                        levelValue="NOR";
                                    }

                                    /******************************日志统计使用**************************************/
                                    Map codeMap=stringRedisTemplate.opsForHash().entries(hashKey);

                                    logger.info("获取数据库值:"+codeMap);
                                    stringRedisTemplate.opsForHash().increment(hashKey,levelValue,1l);
                                    stringRedisTemplate.opsForHash().increment(hashKey,"counter",1l);
                                    if(!codeMap.containsKey("counter")){
                                        stringRedisTemplate.expire(hashKey,13, TimeUnit.HOURS);//设置过期时间
                                    }

                                    logger.error(levelValue+"-----lenvelINfo------>hashKey:"+hashKey);
                                    /**********************************日志统计使用*************************************/



                                }
        }



    }


}
